/* Copyright 203-2115 AtomWare Technology Pty Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * 
 * The flowing information must be keep 
 * RemindEventServiceImpl.java
 * netty-socket-io
 * 2013年9月16日上午10:27:24
 * @author 屈甲康
 * QQ285799123
 * mailto:qujiakang@gmail.com
 */

package org.jackysoft.service.impl;

import io.socket.IOAcknowledge;
import io.socket.IOCallback;
import io.socket.SocketIO;
import io.socket.SocketIOException;

import java.net.MalformedURLException;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.bson.types.ObjectId;
import org.jackysoft.config.ContextConfig;
import org.jackysoft.data.AbstractMongodbDataProvider;
import org.jackysoft.entity.nosql.NotificationMessage;
import org.jackysoft.entity.nosql.RemindEvent;
import org.jackysoft.nosql.UserEvent;
import org.jackysoft.query.RunnerQuery;
import org.jackysoft.service.RemindEventService;
import org.joda.time.DateTime;
import org.json.JSONObject;
import org.mongodb.morphia.query.Query;
import org.mongodb.morphia.query.UpdateOperations;
import org.springframework.util.CollectionUtils;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.gson.Gson;

public class RemindEventServiceImpl extends
AbstractMongodbDataProvider<RemindEvent> implements RemindEventService,IOCallback {

	protected static final Log logger = LogFactory
			.getLog(RemindEventServiceImpl.class);
	@Resource
	protected ContextConfig context;	
	protected static SocketIO socket;
	
	

	public RemindEventServiceImpl() {
		super();
	}

	public void init(){
		try {
			String wsServer = String.format("http://%s:%d/", context.websocketHost,context.websocketPort);
			socket = new SocketIO(wsServer,this);			
			
		} catch (MalformedURLException e) {
			logger.error(e);			
		}
	}
	@Override
	public Collection<RemindEvent> getValidEvents() {
		Query<RemindEvent> query = this.ds.createQuery(type);
		query.filter("valid", true);
		return query.asList();
	}

	@Override
	public void changeValid(boolean valid,String... idz) {
		List<ObjectId> ids = Lists.newArrayList();
		for(String id:idz)ids.add(new ObjectId(id));
		Query<RemindEvent> query = ds.createQuery(type);
	    query.field("id").in(ids);
	    UpdateOperations<RemindEvent> oprs = ds.createUpdateOperations(type);
	    oprs.set("valid", valid);
	    ds.update(query, oprs);
	}

	
	@Override
	public void fireAllValidEvents() throws MalformedURLException {
		Collection<RemindEvent> events = getValidEvents();
		if(CollectionUtils.isEmpty(events))return;
		fireEvents(events);
	}

	@Override
	public void expireEvent(int eventType, String externalKey, String receiver) {
		Query<RemindEvent> query = ds.createQuery(type);
		query.field("receiver").equal(receiver).field("eventType").equal(eventType);
		UpdateOperations<RemindEvent> ops = ds.createUpdateOperations(type).set("valid", false);
		ds.update(query, ops);
	}

	@Override
	public void fireEvents(RunnerQuery query) {
		List<RemindEvent> events = query.mongoQuery(ds, type).asList();
		fireEvents(events);
	}

	Map<String, Collection<NotificationMessage>> convertToNofity(
			Collection<RemindEvent> res) {
		Map<String, Collection<NotificationMessage>> msts = Maps.newHashMap();
		List<String> idz = Lists.newArrayList();
		for (RemindEvent re : res) {
			Collection<NotificationMessage> mt = msts.get(re.getReceiver());
			if (mt == null) {
				mt = Sets.newHashSet();
				msts.put(re.getReceiver(), mt);
			}
			mt.add(new NotificationMessage(re.getId(), re.getContent()));
			idz.add(re.getId());
			
		}		
        changeLastNotified(idz.toArray(new String[]{}));
		return msts;
	}
	void changeLastNotified(String... idz){
		if(idz==null) return;
		long mod = Long.parseLong(DateTime.now().toString("YMd"));		
		List<ObjectId> dl = Lists.newArrayList();
		for(String id:idz)dl.add(new ObjectId(id));		    
	    Query<RemindEvent> query = ds.createQuery(type).field("id").in(dl);
	    UpdateOperations<RemindEvent> ops = ds.createUpdateOperations(type).set("lastNotificed", mod);
	    ds.update(query, ops);
	}

	@Override
	public void fireEvents(Collection<RemindEvent> events){
		Map<String, Collection<NotificationMessage>> msts=convertToNofity(events);
			
		if(socket!=null && socket.isConnected()){	
			
			int page = msts.size()/10;
			int rest = msts.size()%10;
			
			if(page==0){
				sendNotify(msts);
				return;
			}			
		
			Iterator<String> keys = msts.keySet().iterator();		    
		    Map<String,Collection<NotificationMessage>> tmpmsg = Maps.newHashMap();		   
		    for(int start=1;keys.hasNext();start++){
		    	String key = keys.next();
		    	Collection<NotificationMessage> values = msts.get(key);	   
		    	
		    	if(start%10==0){
		    		sendNotify(tmpmsg);	    		
		    	}else{
		    		tmpmsg.put(key, values);
		    	}		    	
		    	
		    }
		    sendNotify(tmpmsg);
    		
			
		}else{
			logger.info("can not connect to websocket server at this memont,reconnect...");			
			init();
		}
	}
	
	void sendNotify(Map<String,Collection<NotificationMessage>>  data){
		Gson gson = new Gson();	
		String str = gson.toJson(data);
		socket.emit(UserEvent.Notified,str);
		logger.info("message event has sent ");
		data.clear();		
	}

	@Override
	public Collection<RemindEvent> getValidEvents(String owner) {
	
		return ds.createQuery(type).filter("receiver", owner).filter("valid", true).asList();
	}

	@Override
	public void changeOneValid(boolean valid, String id) {
		Query<RemindEvent> query = ds.createQuery(type);
	    query.field("id").equal(id);
	    UpdateOperations<RemindEvent> oprs = ds.createUpdateOperations(type);
	    oprs.set("valid", valid);
	    ds.update(query, oprs);
		
	}

	@Override
	public void onDisconnect() {
	 
		
	}

	@Override
	public void onConnect() {
		 logger.info("WebSocket server connect successfully!");
		
	}

	@Override
	public void onMessage(String data, IOAcknowledge ack) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void onMessage(JSONObject json, IOAcknowledge ack) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void on(String event, IOAcknowledge ack, Object... args) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void onError(SocketIOException socketIOException) {
		logger.info("error to connect websocket server ");
		socketIOException.printStackTrace();
		
	}

	

}
